package pub.tbc.api.invoker.protocol.http;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;
import pub.tbc.api.util.NettyHttpUtil;
import pub.tbc.api.service.ServiceInfo;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;

/**
 * @author tbc on 2017/3/10 09:22:46.
 */
@Slf4j
public class NettyHttpInvoker extends AbstractHttpInvoker {
    /**
     * 组合请求uri
     *
     * @param serviceInfo 服务信息
     * @param request     原始请求
     * @return 将要发出的新的uri
     */
    private String outUri(ServiceInfo serviceInfo, FullHttpRequest request) {
        return serviceInfo.getContentPath() + request.uri();
    }

    @Override
    public void invoke(ChannelHandlerContext context, FullHttpRequest request) {
        ServiceInfo serviceInfo = serviceInfo(request);
        String serviceAddress = serviceAddress(serviceInfo);
        InetSocketAddress address = serviceAddress0(serviceAddress);

        Channel inboundChannel = context.channel();

        Bootstrap b = new Bootstrap();
        b.group(NettyHttpUtil.getClientEventLoopGroup())     // use inboundChannel thread
                .channel(context.channel().getClass())
                .handler(new BackendHandlerInitializer(inboundChannel));

        ChannelFuture f = b.connect(address);
        Channel outboundChannel = f.channel();
        request.retain();// 当客户端代码中需要保持一个该对象的引用时，调用此方法将计数加1, 防止release
        ChannelFuture channelFuture = f.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) throws Exception {
                log.debug("[client] 连接建立成功");
                if (future.isSuccess()) {
                    FullHttpRequest req = new DefaultFullHttpRequest(
                            HttpVersion.HTTP_1_1,
                            request.method(),
                            outUri(serviceInfo, request)
                    );
                    req.headers().setAll(request.headers());
                    req.headers().set("Host", serviceAddress);
                    req.headers().set("connection", "keep-alive");
                    log.debug("send request => {}", requestToString(req));

                    outboundChannel.writeAndFlush(req);
                } else {
                    if (!HttpUtil.isKeepAlive(request)) {
                        inboundChannel.close();
                    }
                }
            }
        });
    }


    private static class BackendHandlerInitializer extends ChannelInitializer<SocketChannel> {
        final Channel inboundChannel;

        public BackendHandlerInitializer(Channel inboundChannel) {
            this.inboundChannel = inboundChannel;
        }

        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                    .addLast(new HttpClientCodec())
                    .addLast(new HttpObjectAggregator(1024 * 1024))
                    .addLast(new HttpBackendHandler(inboundChannel));
        }
    }

    @Slf4j
    private static class HttpBackendHandler extends SimpleChannelInboundHandler<FullHttpResponse> {

        private final Channel inboundChannel;

        public HttpBackendHandler(Channel inboundChannel) {
            this.inboundChannel = inboundChannel;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.debug("[out][转发连接]Backend Handler is Active!");
            super.channelActive(ctx);
        }

        @Override
        public void channelRead0(final ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
            log.info("[out]收到第三方响应: {}", msg);
            log.debug("{}", msg.content().toString(StandardCharsets.UTF_8));

            ChannelFuture channelFuture = inboundChannel.writeAndFlush(msg.retain());
            channelFuture.addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture future) throws Exception {
                    log.debug(".......................................");
                    if (!future.isSuccess()) {
                        future.channel().close().sync();
                    }
                }
            });
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            log.debug("[out][连接断开事件]Backend Handler destroyed!");
            NettyHttpUtil.closeOnFlush(inboundChannel);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("[out]异常： {}", cause.getMessage());
            NettyHttpUtil.closeOnFlush(ctx.channel());
        }
    }
}
